Skip to content
Vercel AI SDK 数据流与适配器层
概述
streamText 接收消息和模型，返回 toDataStreamResponse。useChat 发送 POST 请求，拿到 ReadableStream 后逐块解析，更新 UI。
这两个 API 能够跨模型供应商工作，依赖的是两个底层机制：
- Data Stream Protocol——自定义的基于前缀的流协议，在单条 HTTP 连接上承载文本增量、工具调用、元数据和错误。
- LanguageModelV2——所有模型提供者需要遵守的接口契约。
doGenerate负责一次性生成，doStream负责流式生成。上层 API 只依赖这个接口，不感知具体供应商。
先搞清楚协议层和适配器层的分工，再看架构图会更明确：
useChat / useCompletion (客户端状态管理 + 流解析)
─────────────────────────────────────────────
readDataStream (Data Stream Protocol 解析器)
─────────────────────────────────────────────
toDataStreamResponse (协议编码器)
─────────────────────────────────────────────
streamText / generateText / streamObject (核心逻辑)
─────────────────────────────────────────────
LanguageModelV2.doStream / doGenerate (适配器)
─────────────────────────────────────────────
OpenAI / Anthropic / Google / ... 原生 API基本概念
Data Stream Protocol
toDataStreamResponse() 输出的不是 SSE，而是一套以行为单位的编码协议。每一行都有一个类型前缀，后面紧跟一个冒号和 JSON 编码的值。
典型的流内容：
text
0:"Hello"
2:{"tool_calls":[{"type":"function","function":{"name":"get_weather","arguments":"{"city":"Beijing"}"}}]}
d:{"finishReason":"stop"}前缀约定：
0:文本增量（text-delta）2:工具调用（tool-call），值为 JSON 序列化后的数组3:工具执行结果（tool-result）d:结构化元数据，包含finishReason、usage等e:错误信息9:自定义元数据
客户端解析器 readDataStream 根据行首字符将内容分发到不同的回调：遇到 0: 追加到当前助手消息的文本；遇到 2: 更新对话中的工具调用状态；遇到 d: 且 finishReason 为 stop 时标记本轮生成结束。
为什么不是 SSE
SSE 在浏览器端有两个硬性限制。
第一，EventSource 只支持 GET 请求。对话的 messages 数组动辄包含数千 token，放在 URL 查询参数中不可行，而 SSE 没有原生办法走 POST 携带请求体。
第二，EventSource 不支持自定义请求头。需要携带 Authorization 或租户标识的场景无法直接实现。
此外，EventSource 内置自动重连语义。连接断开后是否应该续传、是否应该重新调用模型，这些决策取决于具体业务逻辑，协议层不应该替应用层做决定。
因此 SDK 选择了 fetch + ReadableStream + 基于行的编码，在一条连接上同时传输文本、工具调用和元数据。
LanguageModelV2 接口
所有模型适配器的共同契约定义如下：
ts
interface LanguageModelV2 {
specificationVersion: 'v2';
provider: string;
modelId: string;
doGenerate(options: GenerateOptions): Promise<GenerateResult>;
doStream(options: StreamOptions): Promise<StreamResult>;
}每个供应商（OpenAI、Anthropic、Google 等）分别实现 doGenerate 和 doStream。streamText、generateText、streamObject 等 API 完全不依赖具体供应商的 API 形态。切换 openai('gpt-4o') 到 anthropic('claude-sonnet-4-6') 时，业务代码不需要改一行。
doGenerate 和 doStream 接收的 options 经过了归一化处理，字段包括提示词、最大 token 数、温度、工具定义、工具选择策略等，与供应商无关。
工作原理
readDataStream 的分片拼接
客户端从 fetch 拿到的 response.body 是 ReadableStream<Uint8Array>。网络的 TCP 分片可能在任意位置切断数据，完整的一行 0:"Hello" 可能被拆分到两个 chunk 中。
解析器的核心逻辑以换行符 \n 为分隔，维护一个 buffer 保存不完整的最后一行：
ts
async function readDataStream(
reader: ReadableStreamDefaultReader<Uint8Array>,
callbacks: {
onText?: (text: string) => void;
onToolCall?: (toolCall: ToolCall) => void;
onData?: (data: DataMessage) => void;
onError?: (error: unknown) => void;
}
) {
const decoder = new TextDecoder();
let buffer = '';
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split('\n');
buffer = lines.pop() || '';
for (const line of lines) {
if (!line.trim()) continue;
const type = line[0];
const colonIndex = line.indexOf(':');
const content = line.slice(colonIndex + 1);
switch (type) {
case '0':
callbacks.onText?.(JSON.parse(content));
break;
case '2':
callbacks.onToolCall?.(JSON.parse(content));
break;
case 'd':
callbacks.onData?.(JSON.parse(content));
break;
case 'e':
callbacks.onError?.(JSON.parse(content));
break;
}
}
}
// 处理 buffer 中可能残留的完整行
if (buffer.trim()) {
const type = buffer[0];
const colonIndex = buffer.indexOf(':');
const content = buffer.slice(colonIndex + 1);
callbacks.onData?.(JSON.parse(content));
}
}每一行的内容经过 JSON.parse 还原。如果模型输出中包含换行符，JSON 编码会将其转义为 \n，不会与协议层的行分隔符冲突。
适配器内部的格式归一化
不同供应商的 SSE 格式差异很大。
OpenAI 的流式响应片断：
data: {"choices":[{"delta":{"content":"Hello"}}]}Anthropic 的流式响应片段：
data: {"type":"content_block_delta","delta":{"text":"Hello"}}适配器的工作是把这些异构格式转换为统一的内部类型：
ts
type LanguageModelV2StreamPart =
| { type: 'text-delta'; textDelta: string }
| { type: 'tool-call'; toolCallId: string; toolName: string; args: string }
| { type: 'tool-result'; toolCallId: string; toolName: string; result: unknown }
| { type: 'finish'; finishReason: string; usage: LanguageModelUsage }
| { type: 'error'; error: unknown };streamText 拿到 ReadableStream<LanguageModelV2StreamPart> 后，通过 DataStreamWriter 将 text-delta 映射为 0: 行，tool-call 映射为 2: 行，finish 映射为 d: 行，写入 HTTP 响应流。
useChat 的状态流转
useChat 内部维护了一个状态机，共有四个显式状态和一个隐式状态：
- IDLE——没有正在进行的请求。
- SUBMITTED——
handleSubmit已调用，POST 请求已发出，尚未开始接收响应体。 - STREAMING——正在逐块读取并解析流式响应，UI 随之更新。
- READY——流正常完成，或发生可恢复错误。
隐式的 ABORTING 状态出现在用户快速连续发送消息时：前一个请求尚未完成，新的 handleSubmit 调用会触发 abortController.abort()，然后进入新一轮 SUBMITTED。
useChat 内部的关键变量：
abortControllerRef——当前请求的AbortController引用。messages——会话中所有消息的useState数组。isLoading——是否有未完成的请求。error——最近一次错误对象，不会自动清空。
用户提交消息的完整链路：
- 用户消息和一条空的助手消息被追加到本地
messages数组（乐观更新）。 fetch向/api/chat发送 POST 请求，携带当前messages。- 读取
response.body的 reader，逐块调用readDataStream更新助手消息内容。
连续发送多条消息时，每次新的提交都会 abort 前一个请求的 AbortController。前一个请求已追加到 UI 的文本片段不会被主动清除，但新请求会替换掉最后一条助手消息，因此旧输出会被覆盖而非叠加。
渲染节流
模型通常以每秒 30–80 个 token 的速度产出文本。如果每个 token 到达都触发一次 setState，React 会执行完整的渲染周期。SDK 默认约 50ms 的节流间隔，将连续的 chunk 合并到一次状态更新中，渲染频率控制在每秒 20 次以内。
纯文本更新的 React diff 开销很低，但如果 UI 中包含 Markdown 渲染或代码高亮组件，每次渲染的成本可能明显上升，这种情况下需要在业务层对未变化的消息使用 React.memo 进一步控制。
工具调用的执行模型
工具调用采用声明式模型：开发者定义工具及其 execute 函数，模型决定何时调用哪个工具，SDK 负责执行并将结果传回模型。
带工具的 streamText 内部执行流程：
第 1 轮：model.doStream({ messages, tools })
└─ 流式输出，若 finishReason 为 'tool-calls'，
提取本轮所有 ToolCall。
└─ 如果单轮返回多个互不依赖的 ToolCall，
SDK 并行执行这些工具。
第 2 轮：await Promise.all(toolCalls.map(call => execute(call)))
└─ 构造新的 messages 数组：
[...原消息, assistant(toolCalls), ...toolResults]
└─ model.doStream({ messages: newMessages, tools })
... 直到达到 maxToolRoundtrips 或 finishReason 不再为 'tool-calls'maxToolRoundtrips 默认值为 1，控制工具调用→模型→工具的最大循环轮数。每轮都会向模型 API 发送完整的 messages 数组，包括之前所有轮次的用户消息、助手消息和工具结果。消息越长、轮次越多，token 消耗增长越快。
如果工具在执行期间抛出异常（例如外部 API 超时），streamText 会进入错误路径并中断流。更可控的做法是让工具捕获异常，返回一个包含错误描述的字符串结果，由模型决定如何应对，而不是直接抛异常中断整个流程。
基本用法
安装
bash
npm install ai @ai-sdk/openaiuseChat
tsx
import { useChat } from 'ai/react';
function Chat() {
const { messages, input, handleInputChange, handleSubmit } = useChat({
api: '/api/chat',
});
return (
<div>
<ul>
{messages.map((m, index) => (
<li key={index}>
{m.role === 'user' ? 'User: ' : 'AI: '}
{m.content}
</li>
))}
</ul>
<form onSubmit={handleSubmit}>
<input value={input} onChange={handleInputChange} />
<button type="submit">Send</button>
</form>
</div>
);
}useChat 默认 POST 到 /api/chat。首次调用 handleSubmit 后进入 SUBMITTED 状态，收到响应体后进入 STREAMING 状态，流结束后进入 READY。中途再次提交会 abort 当前请求并开启新一轮 SUBMITTED。
streamText
ts
import { streamText } from 'ai';
import { openai } from '@ai-sdk/openai';
export async function POST(req: Request) {
const { messages } = await req.json();
const result = streamText({
model: openai('gpt-4o'),
messages,
});
return result.toDataStreamResponse();
}streamText 内部创建 DataStreamWriter，将 LanguageModelV2StreamPart 的每个事件映射为协议中带前缀的行，编码后写入 ReadableStream。
定义工具
ts
const result = streamText({
model: openai('gpt-4o'),
messages,
tools: {
get_weather: {
description: '获取指定城市的天气',
parameters: {
type: 'object',
properties: { city: { type: 'string' } },
required: ['city'],
},
execute: async ({ city }) => {
// 实际使用中调用外部天气 API
return `${city} 25°C，晴`;
},
},
},
maxToolRoundtrips: 3,
});当模型返回 finishReason: 'tool-calls' 时，SDK 提取本轮所有工具调用。互不依赖的工具会并行执行。所有工具执行完毕后，结果合并为新的 messages 参与下一轮 doStream。
API
useChat 选项
api——后端端点，默认/api/chat。initialMessages——初始消息数组。body——附加请求体参数。onError——请求或流解析出错时的回调。onFinish——流正常结束时的回调，参数为最终助手消息。experimental_prepareRequestBody——发送请求前动态修改请求体的函数。
返回值：messages、input、handleInputChange、handleSubmit、isLoading、error、reload、stop。
streamText 参数
model——实现了LanguageModelV2的模型实例。messages——对话消息数组。tools——工具定义及execute函数。maxToolRoundtrips——最大工具调用轮次，默认 1。temperature、maxTokens等——标准模型参数。
LanguageModelV2
提供者适配器需要实现的接口，核心方法是 doGenerate 和 doStream，加上 specificationVersion、provider、modelId 三个属性。
注意点
- 流中断后的消息状态：客户端网络断开时，
useChat默认保留已收到的部分文本，不会触发onError。reload()会重新发送最后一条用户消息，导致模型重新生成回复。建议在onFinish中将完整回复持久化，在onError中先查询消息是否已存在，避免重复调用模型。 - 竞态与取消：连续快速发送消息时，SDK 自动 abort 前一个请求，但 UI 上可能短暂残留前一个请求的部分输出。若需要严格避免闪烁，可以在发送新消息时立即清空当前正在流式输出的助手消息。
- 工具调用轮次的成本：每轮工具调用会重新发送完整的消息历史。
maxToolRoundtrips设置得越高，token 消耗和响应延迟增长越快。建议在工具内部加入兜底逻辑（如连续空结果时终止），并监控单轮对话的累计 token 量。 - Edge Runtime 约束：Vercel Edge 存在 30 秒超时和 4 MB 响应体限制。复杂工具调用或多轮对话容易触及这些边界，耗时较长的场景可将繁重工具执行拆分到 Serverless Function 中。
- 不同提供者的行为差异：同一个 system prompt 在 OpenAI 和 Anthropic 上可能产生不同的输出。适配器统一了数据格式，但提示词层面的响应差异需要调用方自行处理。
- 闭包过期问题：
useChat的body属性若依赖外部变量，变量变化后可能不会反映到后续请求中。此时应使用experimental_prepareRequestBody动态计算请求体，或使用 ref 保存最新值。
限制
- 客户端消息数组仅存在于内存中，页面刷新即丢失。SDK 不提供消息持久化、会话管理或分布式部署所需的组件。
- 流式连接是单条 HTTP 连接，无法在多个服务器实例之间迁移。实例宕机会导致流中断，需由客户端重试。
- 工具调用的并行执行限于单轮返回的互不依赖的工具调用。如果模型在第二轮才判断出需要调用另一个工具，该调用仍必须串行等待前一轮完成。多步推理场景下的并行度由模型在单轮中返回的工具调用数量决定。
应用
- 构建流式聊天界面，使用
useChat管理前后端通信和消息状态。 - 集成工具调用，使模型能够查询外部数据或调用外部 API。
- 按照
LanguageModelV2接口实现自定义适配器，接入自部署模型或其他供应商。 - 通过
streamObject实现结构化输出，或通过generateText完成非流式的语义处理。